package cc.verywell.pureblock.net;
/**
 * 
 * 
* 
* Copyright 2019 TanYaqiu
* @author TanYaqiu 
* @address flat 601
* @date 2019年3月9日 下午2:43:18
*/

import java.io.IOException;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import cc.verywell.pureblock.func.DuplicateConnException;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.util.EncryptUtil;

public class NormalChannel extends AbstractChannel{
	final static private int HEADLONG = 4;

	private Context context;
	public Context getContext() {return context;}
	private EncryptUtil eu;
	private int messageSize;
	private NormalFunction callback;
	final NormalFunction getCallback() {return this.callback;}
	private NormalSession session;
	public NormalSession getSession() {return session;}
	private boolean isClient;
	@Override
	public boolean isClient() {return isClient;}
	public NormalChannel(Object own) {
		super(own);
	}
	/**
	 * 就是reset ， NormalChannel必须能复用
	 */
	public NormalChannel init(AsynchronousSocketChannel channel,Context context,EncryptUtil eu,ByteBuffer unread,NormalFunction callback,int salt,boolean isClient) throws IOException {
		
		this.channel = channel;
		this.context = context;
		this.eu = eu;
		this.callback = callback;
		this.isClient = isClient;
		
		if(AioProvider.MESSAGE_SIZE!=this.messageSize) {
			messageSize = AioProvider.MESSAGE_SIZE;
			int size = HEADLONG+eu.outputLength(messageSize);
			this.readBuffer = ByteBuffer.allocate(size);
			this.writeBuffer = ByteBuffer.allocate(size);
		}else {
			this.readBuffer.clear();
			this.writeBuffer.clear();
		}
		{
			////先处理好buffer的lock;
			this.closeState = CloseState.Normal;
			this.lastSize = 0;
			if (this.readLock.availablePermits() != 1) {
				throw new IOException("read@"+this+": readLock not ready!"); //通常发生于内存溢出
			}
			if (this.writeLock.availablePermits() != 1) {
				throw new IOException("write@"+this+": writeLock not ready!"); //通常发生于内存溢出
			}
			List<Msg> writing = writingList;
			if(writing!=null) {
				synchronized (writing) {
					if(writing==this.writingList)this.writeQue=null;
					for (Msg msg : writing) {
						try {
							msg.failed(null, session); //这里也可能发生内存溢出
						} catch (Throwable e1) {
							if(Pblog.IO)Pblog.IO(this,"@write:failed: ",e1);
						}
					}
				}
			}
			this.writeQue.clear();
		}
		writeBuffer.flip(); //否则会一直以为还有数据没写出去;
		session = new NormalSession(this,salt);
		//为了防止 context可能会在 newSession之前收到decode和toDo, 以下会有一些机制;
		boolean hasUnread = (unread!=null && unread.hasRemaining());
		if(readLock.tryAcquire()) {
			boolean readRelease = true;
			try {
				if(hasUnread) {
					readBuffer.put(unread);
				}
				//try {
					//20190317此时可以写，不能读，一切就绪
					this.context.newSession(session);//为了防止 context可能会在 newSession之前收到decode和toDo;
					//newSession出错会退出然后由外边关闭channel,这里不用操作;
				//} catch (Throwable e) {}
				if(hasUnread) {
					readRelease = false;
					readBuffer.flip();
					read(0);
				}
			}finally {
				if(readRelease) {
					readLock.release();
				}
			}
			if(readRelease) {
				startRead();
			}
		}else { ///读线程死锁退出，这个channel不能用了, 外面会new一个的;
			throw new IOException(this+"@read: readLock not ready!"); //not happen;
//			session = null;
//			if(Pblog.IO)Pblog.printIO(this,"@read; ","occupaid.");
//			close(true);
		}
		return this;
	}
	public NormalChannel clearUser() {
		this.channel = null;
		this.context = null;
		this.eu = null;
		this.callback = null;
		this.session = null;
		List<Msg> writing = writingList;
		if(writing!=null) {
			synchronized (writing) {
				if(writing==this.writingList)this.writeQue=null;
				for (Msg msg : writing) {
					try {
						msg.failed(null, session); //这里也可能发生内存溢出
					} catch (Throwable e1) {
						if(Pblog.IO)Pblog.IO(this,"@write:failed: ",e1);
					}
				}
			}
		}
//		if(lastWriteMsg!=null) {
//			try {
//				lastWriteMsg.failed(null, session);
//			} catch (Throwable e1) {
//				if(Pblog.IO)Pblog.IO(this,"@write: ",e1);
//			} finally {
//				lastWriteMsg.clear();
//			}
//		}
		return this;
	}

	
	/**
	 * 写控件
	 * 
	 * 
	 */
	public boolean write(Msg msg) {
		if(closeState==CloseState.Normal) {
			try {
				writeQue.put(msg);
			} catch (InterruptedException e) {
				if(Pblog.IO)Pblog.IO(this,"@write: ",e);
				return false;
			}
			return writeLock_step1();
		}else {
			msg.failed(null, session);
		}
		return false;
	}
	BlockingQueue<Msg> writeQue = new LinkedBlockingQueue<>();
	private List<Msg> writingList = null;
	private ByteBuffer writeBuffer;
	final private Semaphore writeLock = new Semaphore(1); ///用来lock writeBuffer不会给重复调用;
	private boolean writeLock_step1() {
		if(writeLock.tryAcquire()) {
			return writeMsg_step2();
		}else {
			return true;
		}
	}
	private boolean writeMsg_step2() {
		if(writeBuffer.hasRemaining()) {
			writeBuffer_step3();
			return true;
		}
		//下边没有会归null,lastWriteMsg = null;
		Msg firstMsg;
		while(null!=(firstMsg=writeQue.poll())) {
			List<Msg> writing = null;
			try {
				writeBuffer.clear();
				writeBuffer.position(HEADLONG);
				
				Msg nextMsg = writeQue.peek();
				if(nextMsg!=null) {///当前的策略是，灌到满为止，当然也可以选择其它策略的;
					writing = new LinkedList<>();
					writing.add(firstMsg);
					writeBuffer.limit(writeBuffer.position()+messageSize);//锁定limit
					context.encode(firstMsg, session, writeBuffer);
//					if(Pblog.DEBUG)Pblog.DEBUG(this,"@write:multi-write",0,": ",writeBuffer.toString());
					multiwrite:if(writeBuffer.hasRemaining()) {//还能继续写
						do {
							int pos = writeBuffer.position();
							try {
								context.encode(nextMsg, session, writeBuffer);
//								if(Pblog.DEBUG)Pblog.DEBUG(this,"@write:multi-write",writing.size(),": ",writeBuffer.toString());
							}catch(BufferOverflowException boe) {
								writeBuffer.position(pos);
								if(Pblog.DEBUG)Pblog.DEBUG(this,"@write:multi-write-over",writing.size(),": ",writeBuffer.toString());
								break multiwrite;
							}
							writing.add(nextMsg);
							if(nextMsg!=writeQue.poll()) {
								throw new IOException("Msg "+nextMsg+" not at head of que["+writeQue.size()+"]");
							}
						}while(writeBuffer.hasRemaining() && null!=(nextMsg=writeQue.peek()));
					}
				}else {
					writing = Collections.singletonList(firstMsg);
					context.encode(firstMsg, session, writeBuffer);
				}
	
				//这样写就不能使用弹性功能了,所以要在context那边周转一下；msg.write(writeBuffer,session);
				writeBuffer.flip();
				//if(Pblog.DEBUG)Pblog.printDEBUG(this,"@write: remaining ",writeBuffer.remaining());
				writeBuffer.position(HEADLONG);
				int size = eu.encode(writeBuffer);
				if(size==0)throw new IOException("Msg "+firstMsg+"/"+writing.size()+" nothing write to "+session); //会触发msg.failed,这样很好;
				writeBuffer.position(0);
				writeBuffer.putInt(size);
				writeBuffer.position(0);
			} catch (Throwable e) {
				writeBuffer.clear();
				writeBuffer.flip();
				if(Pblog.IO)Pblog.IO(this,"@write: ",e);
				e.printStackTrace();
//				try {
//					lastWriteMsg.failed(e, session);
//					lastWriteMsg = null;
//				} catch (Throwable e1) {
//					if(Pblog.IO)Pblog.IO(this,"@write: ",e);
//				}
				if(writing!=null) {
					synchronized (writing) {
						if(writing==this.writingList)this.writeQue=null;
						for (Msg msg : writing) {
							try {
								msg.failed(null, session); //这里也可能发生内存溢出
							} catch (Throwable e1) {
								if(Pblog.IO)Pblog.IO(this,"@write:failed: ",e1);
							}
						}
					}
				}
			}
			this.writingList = writing; 
			if(writeBuffer.hasRemaining()) {
				writeBuffer_step3();
				return true;
			}
		}
		writeLock.release();
		return false;
	}
	private void writeBuffer_step3() {
		try {
			channel.write(writeBuffer, this, WriteHandler);
		} catch (Throwable e) {
			wroteFailed(e);
		}
	}
	protected void wrote(int result) {
//		if(Pblog.DEBUG)Pblog.DEBUG("wrote ",result," remaining:",writeBuffer.remaining()," ",this);
		writingList = null;
		//if(Pblog.DEBUG)Pblog.printDEBUG(this,"@write: ",result);
		if(session!=null)session.getTimelog().worte(result);
		else if(Pblog.IO)Pblog.IO(this,"@write:failed: wrote session==null, channel suddenly shutdown");//多发生于忽然关闭端口
		if(writeMsg_step2() ) {
			return;
		}
//close那边用线程在等		if(this.closeState==CloseState.Closing) {
//			close(false);
//		}
	}
	protected void wroteFailed(Throwable exc) {
		//有错误，直接关闭连接 readLock.release(); 
		try {
			if(Pblog.IO)Pblog.IO(this,"@write: ",exc);
//			if(lastWriteMsg!=null) {
//				try {
//					lastWriteMsg.failed(exc, session);
//				} catch (Throwable e1) {
//					if(Pblog.IO)Pblog.IO(this,"@write: ",e1);
//				} finally {
//					lastWriteMsg = null;
//				}
//			}
			List<Msg> writing = writingList;
			if(writing!=null) {
				synchronized (writing) {
					if(writing==this.writingList)this.writeQue=null;
					for (Msg msg : writing) {
						try {
							msg.failed(exc, session); //这里也可能发生内存溢出
						} catch (Throwable e1) {
							if(Pblog.IO)Pblog.IO(this,"@write:failed: ",e1);
						}
					}
				}
			}
		} finally {
			writeLock.release();
			close(true);
		}
	}



	
	/**
	 * 读控件
	 * 
	 * 用Semaphore是因为 cc.verywell.pureblock.effictive.ReentreLockTest的性能对比结果
	 * 
	 * startRead()->ReadHandler->readed()->decode->doit->...循环...到最后...startRead()->ReadHandler->readed()->close()
	 * 
	 * 未来方向
	 * 
	 * doit由另外的线程来做，只要doit够一定的数量，就停止继续读下去；当doit低于一定的数量，就恢复startRead(),所以这里先留着readLock;
	 * 
	 * 
	 */
	private int lastSize;
	private ByteBuffer readBuffer;
	private Thread readCustom;
	final private Semaphore readLock = new Semaphore(1); ///用来lock readBuffer不会给重复调用;
	public void startRead() {
		if(/*closeLock==CloseLock.Normal && */readLock.tryAcquire()) { //close后，这里就算继续read, 直到channel.shutdownInput()时也会转到readFailed
			try {
				channel.read(readBuffer,AioProvider.TIME_OUT,TimeUnit.SECONDS,this,ReadHandler);//如果是Server模式有没有都无所谓的。
			} catch (Throwable e) {
				readFailed(e);
			}
		}
	}
	protected void read(int result) {
		//if(Pblog.DEBUG)Pblog.printDEBUG(this,"@read: ",result);
		session.getTimelog().read(result);
		boolean toClose = false;
		try {
			readCustom = Thread.currentThread();
			if(result<0) {//对方通知关闭，直接关闭连接; 
				if(Pblog.DEBUG)Pblog.DEBUG(this,"@read: ","closed.");
				toClose = true;
				return;
			}
			readBuffer.flip();
			try {
				decode:while(readBuffer.hasRemaining()) {
					if(lastSize==0) {
						if(readBuffer.remaining()<4)break decode;
						lastSize = readBuffer.getInt();
						if(lastSize<=0) {
							throw new IOException("read size :"+lastSize+"<=0");
						}
					}
					if(lastSize>readBuffer.remaining()) {
						if(lastSize>readBuffer.capacity()) {
							//可加入自动扩大类 ，或在协议处分包
							throw new IOException("read size :"+lastSize+" > buffer capacity:"+readBuffer.capacity());
						}
						break decode;
					}
					int size = lastSize;
					lastSize = 0;
					int limit = readBuffer.limit();
					int position = readBuffer.position()+size;
					readBuffer.limit(position);
					eu.decode(readBuffer);

					do{
						Msg msg = context.decode(readBuffer,session); // 可能会返回null;
						//if(Pblog.DEBUG)Pblog.DEBUG(this,"@read:multi-read: ",readBuffer.toString());

						try {
							if(msg!=null) { //TODO 未来把msg扔到待办列表就行，别占这个线程; 此时context.toDo别给channel，因为channel要重用，给个小session即可;
								try {
									if(context.toDo(msg, session)) { //由context控制，暂停读包;
										return;
									}
								}catch(Throwable exc) {//执行包错误，报告一下就行，不用关闭连接 ; 
									if(Pblog.IO)Pblog.IO(this,"@doit ",msg,": ",exc);
									exc.printStackTrace();
								}
							}
						}finally {
							if(closeState!=CloseState.Normal) { // 每轮都会最少检查一次 closeState,以防在其间运行了close(); 未来把msg扔到待办列表后，就不用检查了,不是同一个线程;
								toClose = true;
								return;
							}
						}

					}while(readBuffer.hasRemaining());
						
					readBuffer.limit(limit).position(position);
				}
			}catch(Throwable exc) {//解压包有有错误，直接关闭连接 ; 
				if(Pblog.IO)Pblog.IO(this,"@deco: ",exc);
				if(exc.getClass()!=DuplicateConnException.class)
					exc.printStackTrace();
				toClose = true;
				return;
			}
			readBuffer.compact();
		}finally {
			readCustom = null;
			readLock.release(); //先执行解锁最后再重新startRead();
			if(toClose) {
				close(closeState==CloseState.ClosingNow); //默认是false, 但在decode或toDo运行过close(true)就有可能是true;
				return; //需要close就不再startRead()了
			}
		}
		//TODO 未来根据doit待办的数量看看是否startRead();
		startRead();
	}
	protected void readFailed(Throwable exc) {
		//有错误，不执行解锁，直接关闭连接 readLock.release(); 
		try {
			if(Pblog.IO && closeState==CloseState.Normal)Pblog.IO(this,"@read: ",exc);
		}finally {
			readLock.release();
		}
		close(false);
	}
	
	/**
	 * 关闭控件:
	 * 
	 * 
	 */
	private Lock closeLock = new ReentrantLock();
	public CloseState closeState = CloseState.Normal;
	@Override
	public void close(){
		close(true);
	}
	public void close(boolean nowait) {
		if(closeLock.tryLock()) {
			try {
				if(closeState==CloseState.Closed) {
					if(Pblog.IO){
						StackTraceElement stet = Pblog.getSTEwithoutClass(Thread.currentThread().getStackTrace(), this.getClass());
						Pblog.IO(this,"@closed: ","try to close a closed channel from ",stet);
					}
					return;
				}
				if(readCustom==Thread.currentThread()) {
					if(Pblog.DEBUG)Pblog.DEBUG(this,"@read: ","close when reading thread.");
					this.closeState=nowait?CloseState.ClosingNow:CloseState.Closing;
					return;
				}
				//if(session!=null) { //不可能，请报错
					try {
						context.closeSession(session);
					} catch(Throwable e) {
						if(Pblog.INIT)Pblog.INIT(e);
					} finally {
						session.clear();
					} 
				//}
				this.closeState=CloseState.Closed;
				if(Pblog.DEBUG)Pblog.DEBUG(this,"@closing: ","closing...",readLock.availablePermits()==0?"reading...":"",writeLock.availablePermits()==0?"writing...":"");
				if(!nowait && writeLock_step1()) {
					try {
						if(Pblog.IO)Pblog.IO(this,"@closing: ","waiting for write finish.");
						if(writeLock.tryAcquire(AioProvider.CLOSE_TIME_OUT, TimeUnit.SECONDS)) {
							writeLock.release();
						}
					} catch (InterruptedException e) {
						if(Pblog.IO)Pblog.IO(this,"@closing: ",e);
					}
				}
				this.closeState=CloseState.Closed;
				super.close(); //此时随着channel关闭，会触发多个close;
				if(readLock.tryAcquire()) {
					readLock.release();
				}else {
					try {
						if(Pblog.DEBUG)Pblog.DEBUG("! ",this,"@closing: ","waiting for read finish.");
						if(readLock.tryAcquire(AioProvider.CLOSE_TIME_OUT, TimeUnit.SECONDS)) {
							readLock.release();
						}
					} catch (InterruptedException e) {
						if(Pblog.IO)Pblog.IO(this,"@closing: ",e);
					}
				}
				Msg msg;
				while(null!=(msg=writeQue.poll())) {
					try {
						msg.failed(null, session);
					} catch (Throwable e1) {
						if(Pblog.IO)Pblog.IO(this,"@write: ",e1);
					}
				}
			}finally {
				closeLock.unlock();
			}
			if(Pblog.DEBUG)Pblog.DEBUG(this,"@closed: ","closed. ",readLock.availablePermits()==0?"reading...":"",writeLock.availablePermits()==0?"writing...":"");
			callback.closed(this);
		}else if(nowait && closeState!=CloseState.Closed) {
			super.close(); //此时随着channel关闭，会触发多个close;
		}else {
			if(Pblog.DEBUG)Pblog.DEBUG(this,"@closed: ","another close. ");
		}
	}
	enum CloseState{
		Normal,Closing,ClosingNow,Closed;
	}
	
	@Override
	public String toString() {
		NormalSession s = this.getSession();
		if(s==null)
			return super.toString();
		StringBuffer sb = new StringBuffer();
		sb.append(Integer.toHexString(s.salt));
		
		Object herAlias = null;
		if(s.getFunc()!=null)herAlias = s.getFunc().getNode();
		if(herAlias!=null) {
			//sb.append(s.getFunc()==null?null:Integer.toHexString(s.getFunc().hashCode()));
			sb.append("[").append(myAlias()).append(isClient()?"->":"<-");
			sb.append(herAlias);
			AsynchronousSocketChannel channel = getChannel();
			sb.append("(").append(channel!=null?Integer.toHexString(channel.hashCode()):channel).append(")");
			sb.append("]").append(Integer.toHexString(this.hashCode()));
		}else {
			sb.append("<def>");//.append(s.getFunc()==null?null:Integer.toHexString(s.getFunc().hashCode()));
			super.appendNoSessionToString(sb);
		}
		return sb.toString();
	}

}
